package com.udf.common;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

public class BetAccumulatorProcessFunction 
    extends KeyedProcessFunction<Integer, Row, String> {  // Key 类型为 player_id (INT)

    private transient ValueState<Double> totalBetState;

    @Override
    public void open(Configuration parameters) {
        // 定义状态：保存每个玩家的累计投注额
        ValueStateDescriptor<Double> descriptor = 
            new ValueStateDescriptor<>("totalBetState", Double.class);
        totalBetState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(
        Row betRow, 
        Context ctx, 
        Collector<String> out
    ) throws Exception {
        Integer playerId = (Integer) betRow.getField("player_id");
        Double wager = (Double) betRow.getField("casino_win");
        
        // 1. 获取当前累计值（若无则初始化为0）
        Double currentTotal = totalBetState.value() == null ? 0.0 : totalBetState.value();
        currentTotal += wager;
        
        // 2. 更新状态
        totalBetState.update(currentTotal);
        
        // 3. 检查是否达到阈值
        if (currentTotal >= 1000) {
            String alertMsg = String.format(
                "玩家 %d 累计投注已达 %.2f，触发告警！", 
                playerId, currentTotal
            );
            out.collect(alertMsg);
            
            // 可选：触发后重置状态（若需持续监控则删除此行）
            totalBetState.clear();
        }
    }
}